package com.hadoop.sort;


import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class SortReducer extends
Reducer<DateKey, IntWritable, DateKey, IntWritable> {
	

	//reduce 출력키
	private DateKey outputKey = new DateKey();
	//reduce 출력값 
	private IntWritable result = new IntWritable();

	
	//그룹핑 파티셔너와 그룹핑 비교기에 의해 같은 연도로 그룹핑돼 있는 데이터가 전달된 상태 
	//복합키 비교기로 인해 그룹핑된 값은 월의 순서대로 오름차순 정렬되어있음
	//리듀서는 yyyy 그룹키 기준이므로 mm 기준으로 process 진행해야함
	public void reduce(DateKey key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{
		int sum = 0;
		Integer bMonth = key.getMonth();
		System.out.println("Month :: " + bMonth);
		for(IntWritable value : values){
			if(bMonth != key.getMonth()){
				result.set(sum);
				outputKey.setYear(key.getYear());
				outputKey.setMonth(bMonth);
				context.write(outputKey, result);
				sum = 0;
			}
			sum += value.get();
			bMonth = key.getMonth(); 
		}
		
		System.out.println("Last Month :: " + bMonth + " :: " + key.getMonth());
		if(bMonth == key.getMonth()){
			result.set(sum);
			outputKey.setYear(key.getYear());
			outputKey.setMonth(bMonth);
			context.write(outputKey, result);
		}
	}
	
	
}
